/*
 * collector.cc
 *
 *  Created on: Mar 21, 2012
 *      Author: yxjiang
 */

#include "collector.h"

namespace event
{

using namespace std;
using namespace boost::property_tree;

pthread_rwlock_t DataStream::stream_lock_;

DataStream::DataStream(const std::string &name, const string &ip, int port, int time_range_in_sec)
	: stream_ip_(ip), stream_port_(port), time_range_in_sec_(time_range_in_sec)
{
	pthread_rwlock_init(&stream_lock_, NULL);

	name_ = name;
	//	initialize socket configuration
	struct hostent *server_host = NULL;
	server_host = gethostbyname(stream_ip_.c_str());
	bzero((char*)&server_addr_, sizeof(server_addr_));
	server_addr_.sin_family = AF_INET;
	bcopy((char*)server_host->h_addr, (char*) &server_addr_.sin_addr.s_addr, server_host->h_length);
	server_addr_.sin_port = htons(stream_port_);
}

DataStream::~DataStream()
{}

/*
 * Get the name of the stream.
 */
string DataStream::GetName() const
{
	return name_;
}
/*
 * Grab the data from the target monitor,
 * and then put the data into the stream.
 * Remove out-dated data if necessary.
 */
bool DataStream::GrabData()
{
	int grab_socket = socket(AF_INET, SOCK_STREAM, 0);
	if(grab_socket == 0)
	{
		fprintf(stderr, "Create socket failed.\n");
		return false;
	}

	if(connect(grab_socket, (struct sockaddr*)&server_addr_, sizeof(server_addr_)) < 0)
	{
		fprintf(stderr, "Cannot connect to monitor via %s:%d!\n", stream_ip_.c_str(), stream_port_);
		close(grab_socket);
		return false;
	}
	else
	{
		int buffer_size = 4096;
		char in_buf[buffer_size];
		bzero(in_buf, buffer_size);
		pthread_rwlock_wrlock(&stream_lock_);
		stringstream ss;
		while(recv(grab_socket, in_buf, buffer_size, 0) > 0)
		{
			ss << in_buf;
		}
		close(grab_socket);
////		fprintf(stderr, "string: %s\n", ss.str().c_str());
		if(ss.str().length() == 0)
		{
			fprintf(stderr, "Get monitoring meta-data length = 0\n");
			pthread_rwlock_wrlock(&stream_lock_);
			return false;
		}
		SharedPtree tree(new ptree);
		try
		{
			read_json(ss, *tree.get());
		}
		catch (...)
		{
//			fprintf(stderr, "json is not in well structure\n%s\n", ss.str().c_str());
			pthread_rwlock_unlock(&stream_lock_);
			return false;
		}

		time_t cur_time = tree->get<long int>("timestamp");
		//	ignore redundant item
		if(stream_.size() != 0)
		{
			time_t tail_time = stream_.back()->get<long int>("timestamp");
			if(cur_time == tail_time)
			{
				pthread_rwlock_unlock(&stream_lock_);
				return true;
			}
		}
		stream_.push_back(SharedPtree(tree));
		//	remove out-dated items
		time_t head_time = stream_.front()->get<long int>("timestamp");	//	get the head time
		double interval = difftime(cur_time, head_time);
		while(interval > time_range_in_sec_)
		{
			stream_.pop_front();
			head_time = stream_.front()->get<long int>("timestamp");
			interval = difftime(cur_time, head_time);
		}
		pthread_rwlock_unlock(&stream_lock_);
	}
	return true;
}
/*
 * Get a snapshot of the stream.
 */
boost::shared_ptr<std::list<SharedPtree> > DataStream::GetStreamSnapshot()
{
	pthread_rwlock_rdlock(&stream_lock_);
	boost::shared_ptr<list<SharedPtree> > snapshot(new list<SharedPtree>(stream_.begin(), stream_.end()));
	pthread_rwlock_unlock(&stream_lock_);
	return snapshot;
}




map<string, StreamStatus> Collector::streams_;
pthread_mutex_t Collector::streams_lock_;
int Collector::rate_ = 1;
int Collector::buffer_time_length_ = 60;	//	defualt buffer time is 60 seconds
/*
 * Collect the meta-data from the streams.
 */
Collector::Collector(int rate, int registration_port, int buffer_time_len, int retrieve_service_port)
{
	rate_ = rate;
	registration_port_ = registration_port;
	buffer_time_length_ = buffer_time_len;
	retrieve_service_port_ = retrieve_service_port;
	pthread_mutex_init(&streams_lock_, NULL);
}

/*
 *	Start all the services, include all the registered streams
 *	and the retrieve port.
 */
void Collector::Start()
{
	//	start registration service
	int *registration_port = NULL;
	registration_port = &registration_port_;
	pthread_create(&registration_pid_, NULL, _Registration, (void*)registration_port);

	//	start retrieve service
	int *retrieve_port = NULL;
	retrieve_port = &retrieve_service_port_;
	pthread_create(&retrieve_service_pid_, NULL, _RetrieveService, (void*)retrieve_port);

	pthread_join(registration_pid_, NULL);
	pthread_join(retrieve_service_pid_, NULL);
}

/*
 * Handle the monitored node registration.
 */
void *Collector::_Registration(void *arg)
{
	int *pRegistrationPort = (int*)arg;
	struct sockaddr_in server_addr;
	bzero(&server_addr, sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = htons(INADDR_ANY);
	server_addr.sin_port = htons(*pRegistrationPort);

	int registerServerSock = socket(AF_INET, SOCK_STREAM, 0);
	if(registerServerSock < 0)
	{
		fprintf(stderr, "Registration service socket failed!!\n");
		exit(1);
	}

	if(bind(registerServerSock, (struct sockaddr*)&server_addr, sizeof(server_addr)))
	{
		fprintf(stderr, "Registration service bind port: %d failed!!\n", *pRegistrationPort);
		exit(1);
	}

	//	listen
	if(listen(registerServerSock, 5))
	{
		fprintf(stderr, "Registration Service listen failed!!\n");
		exit(1);
	}
	else
	{
		fprintf(stderr, "Wait on port %d for registration...\n", *pRegistrationPort);
	}

	while(true)
	{
		int registrationSock = accept(registerServerSock, NULL, 0);
		stringstream ss;
		char request_buf[1024];
		while(recv(registrationSock, request_buf, 1024, 0) > 0)
		{
			ss << request_buf;
		}
		close(registrationSock);
		vector<string> tokens;
		Split(ss.str(), '\t', tokens);
		//	validate registration information
		if(tokens.size() != 3 || tokens[1].size() == 0 || tokens[2].size() == 0)
		{
			fprintf(stderr, "Registration information incomplete. Received '%s'\n", ss.str().c_str());
			continue;
		}

		//	create the data stream and add to list
		DataStream *stream = new DataStream(tokens[0], tokens[1], atoi(tokens[2].c_str()), buffer_time_length_);
		bool is_success = _RegisterStream(stream);
		if(is_success == false)
		{
			fprintf(stderr, "Register stream failed.\n");
			continue;
		}
		_StartStreams();
		fprintf(stderr, "Monitoring node with name: %s, ip: %s, and port: %s registered.\n",
				tokens[0].c_str(), tokens[1].c_str(), tokens[2].c_str());
	}
	return NULL;
}

/*
 * Register a stream to the collector.
 *  Return Value:
 * 	Return false if a stream with the same name has already been registered.
 * 	Return true if it is successfully registered.
 */
bool Collector::_RegisterStream(DataStream *stream)
{
	string name = stream->GetName();
	pthread_mutex_lock(&streams_lock_);
	map<string, StreamStatus>::const_iterator itr = streams_.find(name);
	if(itr != streams_.end())
		return false;

	StreamStatus status;
	status.pid = 0;
	status.running = false;
	status.stream = stream;
	status.misses = 0;
	streams_.insert(make_pair<string, StreamStatus>(name, status));
	pthread_mutex_unlock(&streams_lock_);
	return true;
}

/*
 * Run each stream as a thread if it is not running.
 */
void Collector::_StartStreams()
{
	pthread_mutex_lock(&streams_lock_);
	map<string, StreamStatus>::iterator itr = streams_.begin();
	for(; itr != streams_.end(); ++itr)
	{
		if(itr->second.running == false)
		{
			pthread_create(&itr->second.pid, NULL, _StreamService, (void*)&itr->second);
			itr->second.running = true;
		}
	}
	pthread_mutex_unlock(&streams_lock_);
}

/*
 * The thread function of the stream.
 * It notify the stream to periodically grab the data.
 */
void *Collector::_StreamService(void *arg)
{
	StreamStatus *status = (StreamStatus*)arg;

	DataStream *stream = status->stream;
	while(true)
	{
		ThreadSleep(rate_, 0);
		bool is_miss = stream->GrabData();
		if(is_miss == false)
		{
			++status->misses;
			fprintf(stderr, "Stream %s missed for %d times.\n", stream->GetName().c_str(), status->misses);
			//	continuous misses for 10 times
			if(status->misses == 30)
			{
				string stream_name = stream->GetName();
				size_t num_erased = streams_.erase(stream_name);
				//	clear all data of this stream
				if(num_erased == 1)
				{
					delete stream;
					fprintf(stderr, "Deregister stream %s.\n", stream_name.c_str());
					pthread_exit(NULL);
				}
				return NULL;
			}

			continue;
		}
		else
		{
//			fprintf(stderr, "Stream '%s' Grab.\n", stream->GetName().c_str());
			status->misses = 0;	//	reset misses once available again
		}
	}
	return NULL;
}

/*
 *	Serve the client's retrieve requests.
 *	Currently, the only service available is to return all the monitored meta-data.
 */
void *Collector::_RetrieveService(void *arg)
{
	int *pRetrievePort = (int*)arg;
	struct sockaddr_in server_addr;	// Server Internet address
	//	initialize server address
	bzero(&server_addr, sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	server_addr.sin_addr.s_addr = htons(INADDR_ANY);
	server_addr.sin_port = htons(*pRetrievePort);

	int retrieveServerSock = socket(AF_INET, SOCK_STREAM, 0);
	if(retrieveServerSock < 0)
	{
		fprintf(stderr, "Retrieve service socket failed!\n");
		exit(1);
	}

	//	bind socket and address
	if(bind(retrieveServerSock, (struct sockaddr*)&server_addr, sizeof(server_addr)))
	{
		fprintf(stderr, "Retrieve service bind port: %d failed!\n", *pRetrievePort);
		close(retrieveServerSock);
		exit(1);
	}
	//	listen
	if(listen(retrieveServerSock, 5))
	{
		fprintf(stderr, "Retrieve service listen failed!\n");
		close(retrieveServerSock);
		exit(1);
	}
	else
		printf("Wait on port %d for Retrieving...\n", *pRetrievePort);

	while(true)
	{
		int retrieve_sock = accept(retrieveServerSock, NULL, 0);
		stringstream ss;
//		char request_buf[1024];
//		while(recv(retrieve_sock, request_buf, 1024, 0) > 0)
//		{
//			ss << request_buf;
//		}
		//	assemble all data streams and return
		ptree root;
		pthread_mutex_lock(&streams_lock_);
		std::map<std::string, StreamStatus>::iterator stream_itr = streams_.begin();
		for(; stream_itr != streams_.end(); ++stream_itr)
		{
			DataStream *stream = stream_itr->second.stream;
			string stream_name = stream->GetName();
			boost::shared_ptr<list<SharedPtree> > snapshot = stream->GetStreamSnapshot();
			if(snapshot->empty() == false)
			{
				SharedPtree last_item = snapshot->back();
				root.add_child(stream_name, *(last_item.get()));	//	only add the latest item
			}
		}
		write_json(ss, root);
		string content = ss.str();
		pthread_mutex_unlock(&streams_lock_);
		const char *send_content = content.c_str();

		if(strlen(send_content) == 0)
		{
			fprintf(stderr, "string len:%ld\n", content.size());
		}

		send(retrieve_sock, send_content, strlen(send_content), 0);
//		fprintf(stderr, "send: %s\n", send_content);
		close(retrieve_sock);
	}
	return NULL;
}

};

int main(int argc, char **argv)
{
	using namespace std;
	using namespace event;

	string ip = "127.0.0.1";
	int port = 32167;

	if(argc >= 2)
		ip = argv[1];
	if(argc >= 3)
		port = atoi(argv[2]);

//	DataStream *stream1 = new DataStream("my-computer", ip, port, 1);

	int rate = 1;
	int registration_port = 32166;
	int retrieve_service_port = 32165;
	int buffer_time_len = 6;	//	set buffer to 60 seconds
	Collector collector(rate, registration_port, buffer_time_len, retrieve_service_port);
//	collector.RegisterStream(stream1);
//	collector._StartStreams();
	collector.Start();

	return 1;
}


